package ink.ykb.mq.config;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import lombok.extern.slf4j.Slf4j;

/**
 * rabbitmq_delayed_message_exchange 插件方式实现
 * 
 * @author mading
 *
 */
@Configuration
@Slf4j
public class DeadDirectRabbitConfig{

	// =====================插件方式实现==============================
	// public static final String DELAY_TEST_QUEUE = "delay_test";
	//
	// public static final String DELAY_TEST_EXCHANGE = "delay_test";

	// //延迟队列
		// @Bean
		// public Queue delayQueue() {
		// return new Queue(DELAY_TEST_QUEUE, true);
		// }
		//
		//// 延迟交换机
		// @Bean
		// public CustomExchange delayExchange() {
		// Map<String, Object> args = new HashMap<>();
		// args.put("x-delayed-type", "direct");
		// return new CustomExchange(DELAY_TEST_EXCHANGE, "x-delayed-message", true,
		// false, args);
		// }
		//
		// //绑定队列到这个延迟交换机上
		// @Bean
		// public Binding binding(Queue delayQueue, CustomExchange delayExchange) {
		// return
		// BindingBuilder.bind(delayQueue).to(delayExchange).with(DELAY_TEST_QUEUE).noargs();
		// }
	
	// =====================TTL和DLX实现==============================
	// 消息延时队列
	public static final String DELAY_PER_QUEUE = "delay_per";

	// 消息延时队列的交换器
	public static final String DELAY_PER_EXCHANGE = "delay_per";

	// 消息过期转发的处理消息的队列
	public static final String DELAY_PROCESS_QUEUE = "delay_process";

	// 消息过期转发的处理消息的交换器
	public static final String DELAY_PROCESS_EXCHANGE = "delay_process";

	
	@Autowired
	  private RabbitTemplateUtil rabbitTemplateUtil;
	
	@Bean
	public RabbitAdmin getRabbitAdmin(ConnectionFactory connectionFactory) {
		RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
		
		
		List<Queue> queueList = getDelayPerQueueList();
		List<Exchange> exchangeList = getDelayPerExchangeList();
		List<Binding> bindingList = getDelayPerBindingList(queueList, exchangeList);
		
		exchangeList.forEach(exchange -> {
			rabbitAdmin.declareExchange(exchange);
		});
		queueList.forEach(queue -> {
			rabbitAdmin.declareQueue(queue);
		});
		bindingList.forEach(binding -> {
			rabbitAdmin.declareBinding(binding);
		});
		

		return rabbitAdmin;
	}
	
	// 消息延时队列
	private List<Queue> getDelayPerQueueList() {
		List<Queue> queueList = new ArrayList<>();
		DelayTime[] values = DelayTime.values();
		for (DelayTime delayTime : values) {
			Queue queue = QueueBuilder.durable(DELAY_PER_QUEUE + "_" + delayTime.name().toLowerCase())
					.withArgument("x-dead-letter-exchange", DELAY_PROCESS_EXCHANGE) // DLX，dead letter发送到的exchange
					.withArgument("x-dead-letter-routing-key", DELAY_PROCESS_QUEUE) // dead letter携带的routing key
					.withArgument("x-message-ttl", delayTime.getTime()).build();
			queueList.add(queue);
		}
		return queueList;

	}

	// 消息延时队列交换机
	private List<Exchange> getDelayPerExchangeList() {
		List<Exchange> exchangeList = new ArrayList<>();
		DelayTime[] values = DelayTime.values();
		for (DelayTime delayTime : values) {
			Exchange exchange = ExchangeBuilder
					.directExchange(DELAY_PER_EXCHANGE + "_" + delayTime.name().toLowerCase()).durable(true).build();
			exchangeList.add(exchange);
		}
		return exchangeList;
	}

	// 绑定消息延时队列交换机
	private List<Binding> getDelayPerBindingList(List<Queue> queueList, List<Exchange> exchangeList) {
		List<Binding> bindingList = new ArrayList<>();

		DelayTime[] values = DelayTime.values();
		for (DelayTime delayTime : values) {
			GO:for (Queue queue : queueList) {
				if (queue.getName().endsWith(delayTime.name().toLowerCase())) {
					for (Exchange exchange : exchangeList) {
						if (exchange.getName().endsWith(delayTime.name().toLowerCase())) {
							Binding binding = BindingBuilder.bind(queue).to(exchange).with(queue.getName()).noargs();
							bindingList.add(binding);
							break GO;
						}
					}
				}
			}
		}

		return bindingList;
	}


	// 消息过期转发的处理消息的队列
	@Bean
	public Queue delayProcessQueue() {
		return new Queue(DELAY_PROCESS_QUEUE, true);
	}

	// 消息过期转发的处理消息的交换器
	@Bean
	public Exchange delayProcessExchange() {
		return ExchangeBuilder.directExchange(DELAY_PROCESS_EXCHANGE).durable(true).build();
	}

	// 绑定交换机
	@Bean
	public Binding delayProcessBinding(Queue delayProcessQueue, Exchange delayProcessExchange) {
		return BindingBuilder.bind(delayProcessQueue).to(delayProcessExchange).with(DELAY_PROCESS_QUEUE).noargs();
	}

//  插件方式实现
//  @RabbitListener(queues= DeadDirectRabbitConfig.DELAY_TEST_QUEUE)
//  TTL和DLX实现
  @RabbitListener(queues= DeadDirectRabbitConfig.DELAY_PROCESS_QUEUE)
  public void delayQueueProcess(Message message) {
    log.info("delayQueueProcess接收消息：{}",new String(message.getBody()));
    MessageProperties messageProperties = message.getMessageProperties();
    Map<String, Object> headers = messageProperties.getHeaders();
    String delayExchange = (String)headers.get("delay-exchange");
    String delayRoutingKey  = (String)headers.get("delay-routing-key");
    rabbitTemplateUtil.sendMsg(delayExchange,delayRoutingKey,message);
    log.info("delayQueueProcess发送消息：{}",new String(message.getBody()));
  }

}
